package com.shujia.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo26PageRank {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("Demo23Cache")

    val sc = new SparkContext(conf)


    //1、读取原始数据
    val dataRDD: RDD[String] = sc.textFile("data/pagerank.txt")

    //2、拆分数据
    val pageLinkRDD: RDD[(String, List[String])] = dataRDD.map(line => {
      val split: Array[String] = line.split("-")
      //当前网页
      val page: String = split(0)
      //出链列表
      val linkList: List[String] = split(1).split(",").toList
      (page, linkList)
    })

    //对多次使用的rdd进行缓存
    pageLinkRDD.cache()


    /**
      * 计算pagerank
      *
      */


    //1、给每一个网页一个初始的pr值 1
    var pagerankRDD: RDD[(String, List[String], Double)] = pageLinkRDD.map {
      case (page: String, linkList: List[String]) =>
        (page, linkList, 1.0)
    }


    var flag = true

    while (flag) {
      //2、将初始的pr值分配给出链列表
      val avgPrRDD: RDD[List[(String, Double)]] = pagerankRDD.map {
        case (page: String, linkList: List[String], pr: Double) =>
          //将pr等分
          val avgPr: Double = pr / linkList.length

          linkList.map(p => (p, avgPr))
      }

      //展开数据
      val pageKVRDD: RDD[(String, Double)] = avgPrRDD.flatMap(list => list)

      //累计每个网页分到的pr值
      val currPrRDD: RDD[(String, Double)] = pageKVRDD.reduceByKey(_ + _)


      //关联出链列表
      val joinRDD: RDD[(String, (Double, List[String]))] = currPrRDD.join(pageLinkRDD)
      //整理数据
      val resultRDD: RDD[(String, List[String], Double)] = joinRDD.map {
        case (page: String, (pr: Double, linkList: List[String])) =>
          (page, linkList, pr)
      }

      /**
        * 计算收敛条件
        *
        */

      //计算网页新的pr值和前一次的pr的差值平均值
      val kvRDD1: RDD[(String, Double)] = resultRDD.map(i => (i._1, i._3))
      val kvRDD2: RDD[(String, Double)] = pagerankRDD.map(i => (i._1, i._3))

      val kvJoinRDD: RDD[(String, (Double, Double))] = kvRDD1.join(kvRDD2)

      //计算差值
      val chaValueRDD: RDD[Double] = kvJoinRDD.map {
        case (_: String, (pr1: Double, pr2: Double)) =>
          Math.abs(pr1 - pr2)
      }

      //差值的平均值
      val chaAvg: Double = chaValueRDD.sum() / chaValueRDD.count()

      //收敛
      if (chaAvg < 0.001) {
        flag = false
      }

      //切换rdd 下一次计算使用新的pr值进行计算
      pagerankRDD = resultRDD

      pagerankRDD.foreach(println)


    }

  }

}
